use std::{
    str::FromStr,
    sync::Arc,
};

use crate::base::{file_name_is_exist, get_sys_date, get_work_path, DiviType, SourceType};
use aho_corasick::{AhoCorasick, PatternID};
use arc_swap::ArcSwapOption;
use chrono::NaiveDate;
use once_cell::sync::Lazy;


use super::sql_parse::sql_item_parse;
pub struct SqlItem {
    pub(super) source: SourceType,
    pub(super) divi_type: DiviType,
    pub(super) name: String,
    pub(super) sql: String,
    pub(crate) col_names: Vec<String>,
    pub(super) sel_col_exps: Vec<String>,
    pub(super) is_preproc: bool,
}
impl SqlItem {
    fn new(
        source: SourceType,
        divi_type: DiviType,
        name: String,
        sql: String,
        col_names: Vec<String>,
        sel_col_exps: Vec<String>,
        is_preproc: bool,
    ) -> Self {
        Self {
            source,
            divi_type,
            name,
            sql,
            col_names,
            sel_col_exps,
            is_preproc,
        }
    }
    pub(super) fn parse(&self) -> anyhow::Result<()> {
        let has_col = self.col_names.iter().any(|c| c == "DTID");
        if !has_col {
            anyhow::bail!("big_table_sql: {} inset中必须含有DTID字段", self.name);
        }
        let has_col = self.col_names.iter().any(|c| c == "EMID");
        if !has_col {
            anyhow::bail!("big_table_sql: {} inset中必须含有EMID字段", self.name);
        }
        let has_col = self.col_names.iter().any(|c| c == "CUID");
        if !has_col {
            anyhow::bail!("big_table_sql: {} inset中必须含有CUID字段", self.name);
        }
        let has_col = self.col_names.iter().any(|c| c == "CYNO");
        if !has_col {
            anyhow::bail!("big_table_sql: {} inset中必须含有CYNO字段", self.name);
        }
        let len1 = self.col_names.len();
        let len2 = self.sel_col_exps.len();
        if len1 != len2 {
            anyhow::bail!("big_table_sql: {} insert字段数{len1}和select字段数{len2}不一致,注意with子句中不能含有select",self.name);
        }
        Ok(())
    }
    pub(super) async fn delete(
        &self,
        clinet: &mut clickhouse_rs::ClientHandle,
        date: &NaiveDate,
        back_where: &str,
    ) -> anyhow::Result<()> {
        self.delete_by_col_names(clinet, date, back_where).await?;
        Ok(())
    }
    // async fn delete_by_sel_col_exps(
    //     &self,
    //     clinet: &mut clickhouse_rs::ClientHandle,
    //     date: &NaiveDate,
    //     back_where: &str,
    //     ac: &AhoCorasick,
    // ) -> anyhow::Result<()> {
    //     let sel_col_exps = self.get_sql_col_exps(false);
    //     let sql = self.sql.replace("#sel_col_exps", &sel_col_exps);
    //     let date_str = date.format("%Y-%m-%d").to_string();
    //     let replace_vals = &[&date_str, back_where];
    //     let mut wtr = vec![];
    //     ac.try_stream_replace_all(sql.as_bytes(), &mut wtr, replace_vals)
    //         .expect("字符串替换失败");
    //     let sql = String::from_utf8(wtr)?;
    //     clinet.execute(sql.as_str()).await?;
    //     Ok(())
    // }
    async fn delete_by_col_names(
        &self,
        clinet: &mut clickhouse_rs::ClientHandle,
        date: &NaiveDate,
        back_where: &str,
    ) -> anyhow::Result<()> {
        let date_str = date.format("%Y-%m-%d").to_string();
        let table = match self.divi_type {
            DiviType::CompBizFee | DiviType::CompAcco => "DFKB_COMP_CLIE_INDEX",
            _ => "DFKB_INDI_CLIE_INDEX",
        };
        let mut ins_cols: Vec<&str> = Vec::new();
        let mut ins_vals: Vec<String> = Vec::new();
        for c in self.col_names.iter() {
            if c == "DTID" || c == "EMID" || c == "CUID" || c == "CYNO" {
                continue;
            } else {
                ins_cols.push(c);
                ins_vals.push(format!("-sum({c}) as {c}"));
            }
        }
        let mut having_vals: Vec<String> = Vec::new();
        for c in ins_cols.iter() {
            having_vals.push(format!("{c} != 0"));
        }
        let sql = format!(
            r#"insert into {table} (DTID,EMID,CUID,CYNO,{}) select DTID,EMID,CUID,CYNO,{}
        from {table} where DTID=toDate('{date_str}') {back_where}
        group by DTID,EMID,CUID,CYNO having {}"#,
            ins_cols.join(","),
            ins_vals.join(","),
            having_vals.join(" or ")
        );
        clinet.execute(sql.as_str()).await?;
        Ok(())
    }
    pub(super) async fn insert(
        &self,
        clinet: &mut clickhouse_rs::ClientHandle,
        date: &NaiveDate,
        back_where: &str,
        ac: &AhoCorasick,
    ) -> anyhow::Result<()> {
        let date_str = date.format("%Y-%m-%d").to_string();
        let replace_vals = &[&date_str, back_where];
        let mut wtr = vec![];
        ac.try_stream_replace_all(self.sql.as_bytes(), &mut wtr, replace_vals)
            .expect("字符串替换失败");
        let sql = String::from_utf8(wtr)?;
        clinet.execute(sql.as_str()).await?;
        Ok(())
    }
    // fn get_sql_col_exps(&self, is_add: bool) -> String {
    //     if is_add {
    //         self.sel_col_exps.join(",")
    //     } else {
    //         let mut res = String::new();
    //         for (i, sce) in self.sel_col_exps.iter().enumerate() {
    //             if let Some(col) = self.col_names.get(i) {
    //                 if !(col == "DTID" || col == "EMID" || col == "CUID" || col == "CYNO") {
    //                     res.push_str("-");
    //                 }
    //             } else {
    //                 continue;
    //             }
    //             res.push_str(sce);
    //             res.push_str(",");
    //         }
    //         res.pop();
    //         res
    //     }
    // }
    pub(super) async fn can_run(&self, date: &NaiveDate) -> anyhow::Result<bool> {
        let edw_date = get_sys_date(SourceType::EDW).await?;
        if self.is_preproc {
            if *date <= edw_date {
                return Ok(false);
            }
        }else{
            if *date > edw_date {
                return Ok(false);
            }
        }
        Ok(true)
    }
}
pub(super) async fn sum_to_empl(
    clinet: &mut clickhouse_rs::ClientHandle,
    divi_type: &DiviType,
    col_names: &Vec<String>,
    date: &NaiveDate,
    and_where: &str,
) -> anyhow::Result<()> {
    let from_table = match divi_type {
        DiviType::CompBizFee | DiviType::CompAcco => "DFKB_COMP_CLIE_INDEX",
        _ => "DFKB_INDI_CLIE_INDEX",
    };
    let mut ins_cols_str = String::new();
    let mut sel_cols_str_for_del = String::new();
    let mut sel_cols_str_for_ins = String::new();
    for col_name in col_names.iter() {
        if col_name == "CUID" || col_name == "CYNO" {
            continue;
        }
        ins_cols_str.push_str(col_name);
        if col_name == "DTID" || col_name == "EMID" {
            sel_cols_str_for_del.push_str(col_name);
            sel_cols_str_for_ins.push_str(col_name);
        } else {
            sel_cols_str_for_del.push_str("-sum(");
            sel_cols_str_for_del.push_str(col_name);
            sel_cols_str_for_del.push_str(")");
            sel_cols_str_for_ins.push_str("sum(");
            sel_cols_str_for_ins.push_str(col_name);
            sel_cols_str_for_ins.push_str(")");
        }
        ins_cols_str.push_str(",");
        sel_cols_str_for_del.push_str(",");
        sel_cols_str_for_ins.push_str(",");
    }
    ins_cols_str.pop();
    sel_cols_str_for_del.pop();
    sel_cols_str_for_ins.pop();
    let date_str = date.format("%Y-%m-%d").to_string();
    //删除旧的数据
    let sql=format!("insert into DFKB_EMPL_INDEX({ins_cols_str}) select {sel_cols_str_for_del} from {from_table} where DTID='{date_str}' {and_where} group by DTID,EMID");
    clinet.execute(sql.as_str()).await?;
    //插入新的数据
    let sql=format!("insert into DFKB_EMPL_INDEX({ins_cols_str}) select {sel_cols_str_for_ins} from {from_table} where DTID='{date_str}' {and_where} group by DTID,EMID");
    clinet.execute(sql.as_str()).await?;
    Ok(())
}

static BIG_TABLE_CONF: Lazy<ArcSwapOption<&'static Vec<SqlItem>>> =
    Lazy::new(|| ArcSwapOption::const_empty());
pub (super) fn init_big_table_sqls() -> anyhow::Result<()> {
    let mut sis = load_big_table_sqls(SourceType::EDW)?;
    verify_sql_items(&sis)?;
    let ods_sis: Vec<SqlItem> = load_big_table_sqls(SourceType::ODS)?;
    verify_sql_items(&ods_sis)?;
    sis.extend(ods_sis);
    BIG_TABLE_CONF.store(Some(Arc::new(Box::leak(Box::new(sis)))));
    Ok(())
}
fn verify_sql_items(items:&Vec<SqlItem>) -> anyhow::Result<()> {
    for si in items.iter() {
        for c in si.col_names.iter() {
            if c == "CUID" || c == "CYNO" || c == "DTID" || c == "EMID" {
                continue;
            }
            for si2 in items.iter() {
                if si2.name == si.name {
                    continue;
                }
                if si2.col_names.contains(c) {
                    return Err(anyhow::anyhow!(
                        "{},{}中存在相同的列{c},这会导致数据清理异常",
                        si.name,
                        si2.name
                    ));
                }
            }
        }
    }
    Ok(())
}
fn load_big_table_sqls(source: SourceType) -> anyhow::Result<Vec<SqlItem>> {
    let file_path = format!("{}/sqls/big_table_{source}.sql", get_work_path());
    if !std::path::Path::new(&file_path).exists() {
        log::warn!("sqls/big_table_{source}.sql文件不存在");
        return Ok(vec![]);
    }
    let sqls_str = std::fs::read_to_string(file_path)?;
    let sqls: Vec<&str> = sqls_str.split(';').collect();
    let mut sis: Vec<SqlItem> = vec![];
    let params_names = &["insert", "select", "from"];
    let pns2 = &["(", ")", ","];
    let ac = AhoCorasick::builder()
        .ascii_case_insensitive(true)
        .build(params_names)
        .unwrap();
    let ac2 = AhoCorasick::builder().build(pns2).unwrap();
    for sql in sqls {
        if sql.trim().is_empty() {
            continue;
        }
        let si = parse_sql(source, sql, &ac, &ac2)?;
        si.parse()?;
        sis.push(si);
    }
    Ok(sis)
}
pub(crate) fn get_big_table_sqls(source: SourceType, name: &str) -> Vec<&'static SqlItem> {
    let sqls = BIG_TABLE_CONF.load();
    let mut ret: Vec<&SqlItem> = vec![];
    if let Some(sqls) = sqls.as_ref() {
        for si in sqls.iter() {
            if si.source == source && si.name == name {
                ret.push(si);
            }
        }
    }
    ret
}
pub fn is_big_table_init() -> bool {
    let sqls = BIG_TABLE_CONF.load();
    sqls.as_ref().is_some()
}
fn parse_sql(
    source: SourceType,
    sql: &str,
    ac: &AhoCorasick,
    ac2: &AhoCorasick,
) -> anyhow::Result<SqlItem> {
    let mut iidx = 0usize;
    let mut sidx = 0usize;
    let mut fidx = 0usize;
    for mat in ac.find_iter(sql) {
        if mat.pattern() == PatternID::must(0) && iidx == 0 {
            iidx = mat.start();
        } else if mat.pattern() == PatternID::must(1) && sidx == 0 {
            let si = mat.start();
            if sql.as_bytes()[si - 1] == b'\n' {
                sidx = si;
            }
        } else if mat.pattern() == PatternID::must(2) && fidx == 0 {
            let fi = mat.start();
            if sql.as_bytes()[fi - 1] == b'\n' {
                fidx = fi;
            }
        }
    }
    if sidx == 0 || fidx == 0 {
        anyhow::bail!("sql:{sql} 语句无效,主句的insert，select，from关键字，必须在行首")
    }
    let input: Vec<char> = sql[0..sidx].chars().collect();
    let (dt, name, is_preproc, col_names) = sql_item_parse()
        .parse(&input)
        .or_else(|err| anyhow::bail!("sql:{sql} 语句无效,{}", err))?;
    if !name.starts_with("ASSE_FTP") && !file_name_is_exist(&name) {
        anyhow::bail!("sql语句的name:{name}在导入配置中不存在")
    }
    let start = sidx + 6;
    let end = fidx;
    let mut sel_col_str = sql[start..end].trim();
    let mut ps_num = 0u8;
    let mut sel_col_exps: Vec<String> = Vec::new();
    let mut prev_idx = 0;
    for p in ac2.find_iter(sel_col_str) {
        if p.pattern() == PatternID::must(0) {
            ps_num += 1;
        } else if p.pattern() == PatternID::must(1) {
            if ps_num == 0 {
                anyhow::bail!("sql:{sql} 语句无效,括号不匹配")
            }
            ps_num -= 1;
        } else {
            if ps_num == 0 {
                let (sp1, sp2) = sel_col_str.split_at(p.start() - prev_idx);
                let col_exp = if prev_idx == 0 {
                    sp1.trim().to_owned()
                } else {
                    sp1[1..].trim().to_owned()
                };
                sel_col_exps.push(col_exp);
                sel_col_str = sp2;
                prev_idx = p.start();
            }
        }
    }
    if sel_col_str.len() > 0 {
        sel_col_exps.push(sel_col_str[1..].to_owned());
    }
    let divi_type = DiviType::from_str(&dt).or_else(|err| {
        anyhow::bail!("bigtable中的sql语句{name}中的divi_type:{dt}配置错误:{err}")
    })?;
    let sql = format!(
        "{} {} {}",
        &sql[iidx..start],
        sel_col_exps.join(","),
        &sql[end..]
    );
    let si = SqlItem::new(
        source,
        divi_type,
        name,
        sql,
        col_names,
        sel_col_exps,
        is_preproc,
    );
    Ok(si)
}
